package cn.mesmile.flink.window;

import cn.mesmile.flink.jdkstream.VideoOrder;
import cn.mesmile.flink.source.VideoOrderSourceV2;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.IntSummaryStatistics;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author zb
 * @date 2021/8/23 22:46
 * @Description  全量窗口函数
 */
public class FlinkApplyWindowsApp {

    /**
     * apply(new WindowFunction(){ })
     *
     * - 【窗口先缓存该窗口所有元素】，等窗口的全部数据收集起来后再触发条件计算
     * - 常见的全窗口聚合函数 windowFunction(未来可能弃用)、processWindowFunction(可以获取到窗口上下文 更多信息,包括窗口信息)
     * IN是输入类型，OUT是输出类型，KEY是分组类型，W是时间窗
     * WindowFunction<IN, OUT, KEY, W extends Window>
     */
    public static void main(String[] args) throws Exception {

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 设置成 单个线程运行 方便查询演示效果
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStreamSource<VideoOrder> ds = env.addSource(new VideoOrderSourceV2());

        KeyedStream<VideoOrder, String> keyByDb = ds.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder videoOrder) throws Exception {
                return videoOrder.getTitle();
            }
        });

        SingleOutputStreamOperator<VideoOrder> apply = keyByDb.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply(new WindowFunction<VideoOrder, VideoOrder, String, TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow timeWindow, Iterable<VideoOrder> input, Collector<VideoOrder> out) throws Exception {
                        List<VideoOrder> list = IteratorUtils.toList(input.iterator());
                        int total = list.stream().collect(Collectors.summingInt(VideoOrder::getMoney)).intValue();
                        VideoOrder videoOrder = new VideoOrder();
                        videoOrder.setMoney(total);
                        videoOrder.setCreateTime(list.get(0).getCreateTime());
                        videoOrder.setTitle(list.get(0).getTitle());

                        out.collect(videoOrder);
                    }
                });


        apply.print("result:");

        env.execute("count windows job");
    }
}
